package com.socket.web;

import java.io.BufferedWriter;
import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStreamWriter;
import java.net.Socket;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

public class SocketManager extends Thread {

    public static final int SOCKET_RECEIVE = 0x0099;
    public static final int SOCKET_CONNECT_SUCCESS = 0x0098;
    public static final int SOCKET_UNEXPECTED_CLOSE = 0x0097;
    public static final int SOCKET_CLOSE_OK = 0x0096;
    public static final int SOCKET_CLOSE_ERR = 0x0095;

    private static final String CHARSET_NAME = "UTF-8";

    private boolean noTask, unexpectedClose, isConnected;
    private Socket mSocket;
    private SocketCallback mCallback;
    private BufferedWriter bw = null;
    private int what = 0;
    private BlockingQueue<String> msgPool;
    
    private FileOutputStream fos;
    private SimpleDateFormat sdf;

    // 在构造中得到要单独会话的socket
    public SocketManager(Socket socket, SocketCallback callback) {
        this.mSocket = socket;
        this.mCallback = callback;
        msgPool = new LinkedBlockingQueue<>();
        sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss:SSS");
        // 返回给客户端数据
        try {
            bw = new BufferedWriter(new OutputStreamWriter(socket.getOutputStream(), CHARSET_NAME));
            FileUtil.createDirFile("/temp");
        	File file = FileUtil.createFile("/temp/temp.csv");
        	fos = new FileOutputStream(file, true);
        } catch (IOException e) {
            e.printStackTrace();
        }
        noTask = true;
        isConnected = true;
        sendMessage(SOCKET_CONNECT_SUCCESS, socket);
    }

    /**
     * 发送消息
     *
     * @param s
     */
    public void send(String s) {
        if (!isConnected)
            return;
        msgPool.add(s);
        what = 1;
        noTask = false;
        synchronized (this) {
            notify();
        }
    }

    /**
     * 关闭连接
     */
    public void close() {
        if (mSocket.isClosed()) {
        	mCallback = null;
            if (msgPool != null) {
                msgPool.clear();
            }
            return;
        }
        what = -1;
        noTask = false;
        synchronized (this) {
            notify();
        }
    }

    /**
     * 是否断开连接
     *
     * @return
     */
    public boolean isClosed() {
        return !isConnected || mSocket.isClosed();
    }

    @Override
    public void run() {
        super.run();
        try {
        	InputStream is = mSocket.getInputStream();
            new Listener(is).start();
        } catch (IOException e) {
            e.printStackTrace();
        }

        while (!mSocket.isClosed()) {
            synchronized (this) {
                while (noTask) {
                    try {
                        wait();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
            noTask = true;
            if (what == 1) {
                try {
                    String msg = msgPool.poll(30, TimeUnit.MINUTES);
                    bw.write(msg + "\n");
                    bw.flush();
                } catch (Exception e) {
                    e.printStackTrace();
                    unexpectedClose();
                }
            } else if (what == -1) {
                try {
                    mSocket.close();
                    if (!unexpectedClose)
                        sendEmptyMessage(SOCKET_CLOSE_OK);
                } catch (IOException e) {
                    e.printStackTrace();
                    if (!unexpectedClose)
                        sendEmptyMessage(SOCKET_CLOSE_ERR);
                }
                isConnected = false;
                mCallback = null;
                if (msgPool != null) {
                    msgPool.clear();
                }
            }
        }
    }

    /**
     * 意外断开
     */
    private void unexpectedClose() {
        unexpectedClose = true;
        close();
        sendEmptyMessage(SOCKET_UNEXPECTED_CLOSE);
    }

    /**
     * 发送消息
     *
     * @param what
     * @param obj
     */
    private void sendMessage(int what, Object obj) {
        if (mCallback != null) {
            mCallback.handleMessage(what, obj);
        }
    }

    /**
     * 发送空消息
     *
     * @param what
     */
    private void sendEmptyMessage(int what) {
    	sendMessage(what, null);
    }

    private class Listener extends Thread {
        private InputStream is;

        public Listener(InputStream is) {
            this.is = is;
        }

        @Override
        public void run() {
            super.run();
            try {
            	byte[] bs = new byte[1024 * 1024];
                int len = 0;
                while ((len = is.read(bs)) != -1) {
                	fos.write(",".getBytes());
                	fos.write(sdf.format(new Date()).getBytes());
                	fos.write((" " + len + " ").getBytes());
                	fos.write(bs, 0, len);
                    sendMessage(SOCKET_RECEIVE, new String(bs, 0, len));
                    send("保存成功");
                }
                // 由远程关闭引发
                unexpectedClose();
            } catch (IOException e) {   // 由socket主动关闭或网络错误引发
                e.printStackTrace();
                unexpectedClose();
            }
        }
    }
    
    public interface SocketCallback {
    	void handleMessage(int what, Object obj);
    }
}
